package com.zxj.hadoop.demo.mapreduce.staffspend.groupuser;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * @Author 朱小杰
 * 时间 2017-07-29 .15:48
 * 说明 带有排序功能的统计，
 */
public class SortGroupUser {

    public static class SortMapper extends Mapper<LongWritable,Text,Spend,Text>{
        private Spend spend = new Spend();
        private IntWritable moneyWritable = new IntWritable();
        private Text text = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split("\t");//这里就不做异常处理了，只写核心逻辑
            String name = split[0];
            int money = Integer.parseInt(split[1]);
            text.set(name);
            moneyWritable.set(money);
            spend.set(text, moneyWritable);
            context.write(spend,text);
        }
    }

    public static class SortReducer extends Reducer<Spend,Text,Text,Spend>{
        /**
         * 因为在这之前已经是汇总的结果了，所以这里直接输出就行了
         * @param key
         * @param values  这里面只有一个，就是姓名
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Spend key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            context.write(values.iterator().next(),key);
        }
    }


    /**
     * 我们需要实现一个新的接口，这个接口包含了排序接口以及序列化接口
     */
    public static class Spend implements WritableComparable<Spend>{
        private Text name; //姓名
        private IntWritable money; //花费

        public Spend(){}

        public Spend(Text name, IntWritable money) {
            this.name = name;
            this.money = money;
        }

        public void set(Text name, IntWritable money) {
            this.name = name;
            this.money = money;
        }
        @Override
        public int compareTo(Spend o) {
            return o.getMoney().get() - this.money.get();
        }

        @Override
        public void write(DataOutput out) throws IOException {
            name.write(out);
            money.write(out);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            name = new Text();
            name.readFields(in);
            money = new IntWritable();
            money.readFields(in);
        }


        public Text getName() {
            return name;
        }

        public void setName(Text name) {
            this.name = name;
        }

        public IntWritable getMoney() {
            return money;
        }

        public void setMoney(IntWritable money) {
            this.money = money;
        }

        @Override
        public String toString() {
            return name.toString() + "\t" + money.get();
        }
    }


    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration config = new Configuration();

        Job job = Job.getInstance(config);

        job.setJarByClass(SortGroupUser.class);

        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);

        job.setMapOutputKeyClass(Spend.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Spend.class);

        FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        boolean b = job.waitForCompletion(true);
        if(b){
            //success
        }

    }
}
